﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using C5;
using Flower.Actors;
using Flower.Logging;
using Flower.Directory.Util;
using Flower.Security;
using Flower.Services;
using Flower.Services.Data;

namespace Flower.Directory.Host.Default
{
    public class ProcessesDispatcher : IProcessesDispatcher
    {
        private static readonly Log Log = LogManager.CreateLog(typeof(ProcessesDispatcher).Name);

        #region Commands

        private interface ICommand : IMessage<ProcessesDispatcher>
        {
            int Priority { get; }
        }

        private class CommandComparer : IComparer<ICommand>
        {
            public static readonly CommandComparer Instance = new CommandComparer();

            public int Compare(ICommand x, ICommand y)
            {
                return x.Priority - y.Priority;
            }
        }

        private class InitCommand : ICommand
        {
            public int Priority
            {
                get { return 0; }
            }

            public void Process(ProcessesDispatcher processor)
            {
                processor.OnInit(this);
            }
        }

        private class RefreshProcessorCommand : ICommand
        {
            public string ProcessorName;

            public int Priority
            {
                get { return 1; }
            }

            public void Process(ProcessesDispatcher processor)
            {
                processor.OnRefreshProcessor(this);
            }
        }

        private class RefreshWorkflowCommand : ICommand
        {
            public string Workflow;

            public int Priority
            {
                get { return 1; }
            }

            public void Process(ProcessesDispatcher processor)
            {
                processor.OnRefreshWorkflow(this);
            }
        }

        private class ProcessStatusChangedCommand : ICommand
        {
            public string ProcessorName;
            public ProcessStatus NewStatus;

            public int Priority
            {
                get { return 1; }
            }

            public void Process(ProcessesDispatcher processor)
            {
                processor.OnProcessStatusChanged(this);
            }
        }

        private class SelectProcessorCommand : ICommand
        {
            public string Workflow;
            public BlockingCollection<string> ProcessorName;

            public int Priority
            {
                get { return 2; }
            }

            public void Process(ProcessesDispatcher processor)
            {
                processor.OnSelectProcessor(this);
            }
        }

        #endregion

        #region Data

        private class ProcessorInfo
        {
            public IPriorityQueueHandle<ProcessorInfo> Handle;
            public string Name;
            public long PendingCount;
            public long RunningCount;

            public long TotalLoad { get { return PendingCount + RunningCount; } }
        }

        private class ProcessorInfoComparer : IComparer<ProcessorInfo>
        {
            public static readonly ProcessorInfoComparer Instance = new ProcessorInfoComparer();

            public int Compare(ProcessorInfo x, ProcessorInfo y)
            {
                return (int)(x.TotalLoad - y.TotalLoad);
            }
        }

        private readonly Actor<ProcessesDispatcher, ICommand> _actor;  

        private readonly Dictionary<string, ProcessorInfo> _processorsByName =
            new Dictionary<string,ProcessorInfo>();
        private readonly Dictionary<string, C5.HashSet<string>> _workflowsPreferredProcessors =
            new Dictionary<string, C5.HashSet<string>>();
        private readonly IntervalHeap<ProcessorInfo> _processorsByPriority =
            new IntervalHeap<ProcessorInfo>(ProcessorInfoComparer.Instance);

        private volatile IDirectory _directory;

        #endregion

        #region Constructor

        public ProcessesDispatcher()
        {
            _actor = new Actor<ProcessesDispatcher, ICommand>
            (
                this,
                new PriorityQueue<ICommand>(CommandComparer.Instance)
            );
            _actor.Error += (sender, ev) =>
                Log.Error
                (
                    string.Format
                    (
                        "Error occurred while processing the command {0}.",
                        ev.Message.GetType().Name
                    ),
                    ev.Exception
                );
        }

        #endregion

        #region IProcessesDispatcher

        public IDirectory Directory
        {
            set
            {
                _directory = value;
                _actor.Send(new InitCommand());
            }
        }

        public string SelectProcessor(string workflow)
        {
            if (_directory == null) throw new InvalidOperationException("Directory is not set");

            var result = new BlockingCollection<string>(1);

            _actor.Send
            (
                new SelectProcessorCommand 
                {
                    Workflow = workflow,
                    ProcessorName = result
                }
            );

            return result.Take();
        }

        public void RefreshProcessor(string processorName)
        {
            if (_directory == null) throw new InvalidOperationException("Directory is not set");

            _actor.Send(new RefreshProcessorCommand { ProcessorName = processorName });
        }

        public void RefreshWorkflow(string workflow)
        {
            if (_directory == null) throw new InvalidOperationException("Directory is not set");

            _actor.Send(new RefreshWorkflowCommand { Workflow = workflow });
        }

        public void ProcessStatusChanged(ProcessStatus newStatus, string processorName)
        {
            if (_directory == null) throw new InvalidOperationException("Directory is not set");

            _actor.Send(new ProcessStatusChangedCommand { ProcessorName = processorName, NewStatus = newStatus });
        }

        #endregion

        #region Command handers

        private void RegisterProcessor(string processorName)
        {
            var processorInfo = new ProcessorInfo
            {
                Name = processorName
            };

            _processorsByName[processorInfo.Name] = processorInfo;

            processorInfo.PendingCount =
                _directory.CountChildren
                (
                    string.Concat("/Processors/", processorInfo.Name, "/Pending"),
                    DirectoryEntryTypes.Processor
                );
            processorInfo.PendingCount =
                _directory.CountChildren
                (
                    string.Concat("/Processors/", processorInfo.Name, "/Running"),
                    DirectoryEntryTypes.Processor
                );

            IPriorityQueueHandle<ProcessorInfo> handle = null;
            _processorsByPriority.Add(ref handle, processorInfo);
            processorInfo.Handle = handle;

            Log.Trace
            (
                () => 
                    string.Format
                    (
                        "Processor {0} registered, pending = {1}, running = {2}.", 
                        processorInfo.Name,
                        processorInfo.PendingCount,
                        processorInfo.RunningCount
                    )
            );
        }

        private C5.HashSet<string> GetWorkflowPreferredProcessors(string workflow)
        {
            using (new AuthorizationOverride("Administrator"))
            {
                C5.HashSet<string> preferredProcessors = null;

                if (_workflowsPreferredProcessors.TryGetValue(workflow, out preferredProcessors))
                {
                    Log.Trace(() => string.Format("Workflow {0} found in the cache.", workflow));
                    return preferredProcessors;
                }

                var wf = _directory.Get(workflow) as Workflow;

                if (wf == null)
                {
                    Log.Trace(() => string.Format("Workflow {0} is not found in the directory.", workflow));
                    return null;
                }

                Log.Trace(() => string.Format("Workflow {0} loaded from the directory.", workflow));

                if (wf.PreferredProcessors.Count > 0)
                {
                    preferredProcessors = new C5.HashSet<string>();
                    preferredProcessors.AddAll(wf.PreferredProcessors);
                }

                _workflowsPreferredProcessors[workflow] = preferredProcessors;

                return preferredProcessors;
            }
        }

        private void OnInit(InitCommand command)
        {
            using (new AuthorizationOverride("Administrator"))
            {
                var activeProcesses =
                    _directory.GetChildrenInfo
                    (
                        "/Processors/Active",
                        DirectoryEntryTypes.Link,
                        SortProperty.Default,
                        SortOrder.Asc,
                        0,
                        long.MaxValue
                    );

                Log.Trace(() => string.Format("Found {0} active processors.", activeProcesses.Count));

                foreach (DirectoryEntryInfo processorLink in activeProcesses)
                {
                    RegisterProcessor(processorLink.Name);
                }
            }
        }

        private void OnSelectProcessor(SelectProcessorCommand command)
        {
            try
            {
                if (_processorsByPriority.Count == 0)
                {
                    command.ProcessorName.Add(null);
                }

                ProcessorInfo processorInfo = _processorsByPriority.FindMin();

                C5.HashSet<string> preferredProcessors = GetWorkflowPreferredProcessors(command.Workflow);

                if (preferredProcessors != null)
                {
                    processorInfo = null;

                    foreach (ProcessorInfo pInf in _processorsByPriority)
                    {
                        if (preferredProcessors.Contains(pInf.Name))
                        {
                            processorInfo = pInf;
                            break;
                        }
                    }
                }

                command.ProcessorName.Add(processorInfo != null ? processorInfo.Name : null);
            }
            catch(Exception)
            {
                command.ProcessorName.Add(null);
            }
        }

        private void OnRefreshProcessor(RefreshProcessorCommand command)
        {
            ProcessorInfo processorInfo;

            if (_processorsByName.TryGetValue(command.ProcessorName, out processorInfo))
            {
                using (new AuthorizationOverride("Administrator"))
                {
                    if (_directory.Exists("/Processors/Active/" + command.ProcessorName, DateTime.MinValue))
                    {
                        _processorsByPriority.Delete(processorInfo.Handle);

                        processorInfo.PendingCount =
                            _directory.CountChildren
                            (
                                string.Concat("/Processors/", processorInfo.Name, "/Pending"),
                                DirectoryEntryTypes.Processor
                            );
                        processorInfo.PendingCount =
                            _directory.CountChildren
                            (
                                string.Concat("/Processors/", processorInfo.Name, "/Running"),
                                DirectoryEntryTypes.Processor
                            );

                        _processorsByPriority.Add(ref processorInfo.Handle, processorInfo);
                    }
                    else
                    {
                        _processorsByName.Remove(processorInfo.Name);
                        _processorsByPriority.Delete(processorInfo.Handle);
                    }
                }
            }
            else
            {
                using (new AuthorizationOverride("Administrator"))
                {
                    if (_directory.Exists("/Processors/Active/" + command.ProcessorName, DateTime.MinValue))
                    {
                        RegisterProcessor(command.ProcessorName);
                    }
                }
            }
        }

        private void OnProcessStatusChanged(ProcessStatusChangedCommand command)
        {
            ProcessorInfo processorInfo;

            if (_processorsByName.TryGetValue(command.ProcessorName, out processorInfo))
            {
                _processorsByPriority.Delete(processorInfo.Handle);

                switch (command.NewStatus)
                {
                    case ProcessStatus.Pending:
                        processorInfo.PendingCount++;
                        break;
                    case ProcessStatus.Running:
                        processorInfo.PendingCount--;
                        processorInfo.RunningCount++;
                        break;
                    case ProcessStatus.Waiting:
                    case ProcessStatus.Finished:
                    case ProcessStatus.Broken:
                        processorInfo.RunningCount--;
                        break;
                }

                _processorsByPriority.Add(ref processorInfo.Handle, processorInfo);
            }
        }

        private void OnRefreshWorkflow(RefreshWorkflowCommand command)
        {
            _workflowsPreferredProcessors.Remove(command.Workflow);
        }

        #endregion
    }
}
